﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace ChinaScope.DAS.Common
{
    public interface IBJKafkaClient
    {
        Task RunConsumeAsync(string brokerList, List<string> topics, Func<ConsumeResult<Ignore, string>, Task> func, CancellationToken cancellationToken, string groupId=null);
        Task<DeliveryResult<Null, string>> ProduceAsync(string brokerList, string topicName, string message);
        void BatchProduce(string brokerList, string topicName, List<string> messages);
    }
    public class BJKafkaClient : IBJKafkaClient
    {
        public void BatchProduce(string brokerList, string topicName, List<string> messages)
        {
            var conf = new ProducerConfig { BootstrapServers = brokerList };

            static void handler(DeliveryReport<Null, string> r) =>
                Console.WriteLine(!r.Error.IsError
                    ? $"Delivered message to {r.TopicPartitionOffset}"
                    : $"Delivery Error: {r.Error.Reason}");

            using var p = new ProducerBuilder<Null, string>(conf).Build();
            messages.ForEach(m =>
{
p.Produce(topicName, new Message<Null, string> { Value = m }, handler);
});
            // wait for up to 10 seconds for any inflight messages to be delivered.
            p.Flush(TimeSpan.FromSeconds(10));
        }

        public async Task<DeliveryResult<Null, string>> ProduceAsync(string brokerList, string topicName, string message)
        {
            DeliveryResult<Null, string> result=null;
            var config = new ProducerConfig { BootstrapServers = brokerList };
            using (var producer = new ProducerBuilder<Null, string>(config).Build())
            {
                Console.WriteLine("\n-----------------------------------------------------------------------");
                Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}.");
                Console.WriteLine("-----------------------------------------------------------------------");
                try
                {
                    // Note: Awaiting the asynchronous produce request below prevents flow of execution
                    // from proceeding until the acknowledgement from the broker is received (at the 
                    // expense of low throughput).
                    result = await producer.ProduceAsync(
                        topicName, new Message<Null, string> { Value = message });
                    Console.WriteLine($"delivered to: {result.TopicPartitionOffset}");
                }
                catch (ProduceException<string, string> e)
                {
                    Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}]");
                }
                // Since we are producing synchronously, at this point there will be no messages
                // in-flight and no delivery reports waiting to be acknowledged, so there is no
                // need to call producer.Flush before disposing the producer.
            }
            return result;
        }

        public async Task RunConsumeAsync(string brokerList, List<string> topics, Func<ConsumeResult<Ignore, string>, Task> func, CancellationToken cancellationToken, string groupId = null)
        {
            var config = new ConsumerConfig
            {
                BootstrapServers = brokerList,
                GroupId = groupId ?? Guid.NewGuid().ToString(),
                EnableAutoCommit=true,
                AutoOffsetReset = AutoOffsetReset.Earliest
            };
            using var consumer = new ConsumerBuilder<Ignore, string>(config)
                // Note: All handlers are called on the main .Consume thread.
                .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
                .SetStatisticsHandler((_, json) => Console.WriteLine($"Statistics: {json}"))
                .SetPartitionsAssignedHandler((c, partitions) =>
                {
                    Console.WriteLine($"Assigned partitions: [{string.Join(", ", partitions)}]");
                    // possibly manually specify start offsets or override the partition assignment provided by
                    // the consumer group by returning a list of topic/partition/offsets to assign to, e.g.:
                    // 
                    // return partitions.Select(tp => new TopicPartitionOffset(tp, externalOffsets[tp]));
                })
                .SetPartitionsRevokedHandler((c, partitions) =>
                {
                    Console.WriteLine($"Revoking assignment: [{string.Join(", ", partitions)}]");
                })
                .Build();
            consumer.Subscribe(topics);

            try
            {
                while (true)
                {
                    try
                    {
                        var consumeResult = consumer.Consume(cancellationToken);

                        if (consumeResult.IsPartitionEOF)
                        {
                            Console.WriteLine(
                                $"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");

                            continue;
                        }

                        Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Value}");
                        await func(consumeResult);
                    }
                    catch (ConsumeException e)
                    {
                        Console.WriteLine($"Consume error: {e.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {
                Console.WriteLine("Closing consumer.");
                consumer.Close();
                throw;
            }
        }
    }
}
